package com.zhang.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.sun.xml.internal.bind.v2.TODO;
import com.zhang.gmall.realtime.beans.VisitorStats;
import com.zhang.gmall.realtime.common.GmallConfig;
import com.zhang.gmall.realtime.utils.ClickHouseUtil;
import com.zhang.gmall.realtime.utils.DateTimeUtil;
import com.zhang.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.time.Duration;
import java.util.Date;

/**
 * @title: 访客主题统计
 * @author: zhang
 * @date: 2022/3/13 14:55
 */
public class VisitorStatsApp {
    public static void main(String[] args) throws Exception {
        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        //TODO 2.设置检查点
        //TODO 3.从kafka读取dwd_page_log_2022主题数据、转化为VisitorStats pojo类、分配Watermark和提取事件时间戳
        String pageLogTopic = "dwd_page_log_2022";
        String groupId = "visitor-stats-app";
        SingleOutputStreamOperator<VisitorStats> pageStatsDS = env
                .addSource(MyKafkaUtil.getKafkaSource(pageLogTopic, groupId))
                .map(new MapFunction<String, VisitorStats>() {
                    @Override
                    public VisitorStats map(String value) throws Exception {
                        JSONObject jsonObject = JSON.parseObject(value);
                        JSONObject common = jsonObject.getJSONObject("common");
                        JSONObject page = jsonObject.getJSONObject("page");
                        String lastPageId = page.getString("last_page_id");
                        long sv = lastPageId == null || lastPageId.length() <= 0 ? 1L : 0L;
                        VisitorStats visitorStats = new VisitorStats(
                                "",
                                "",
                                common.getString("vc"),
                                common.getString("ch"),
                                common.getString("ar"),
                                common.getString("is_new"),
                                0L, 1L, sv, 0L, page.getLong("during_time"), jsonObject.getLong("ts")
                        );
                        return visitorStats;
                    }
                });
        //TODO 4.从kafka读取dwm_unique_visit_2022主题数据、转化为VisitorStats pojo类、分配Watermark和提取事件时间戳
        String uvTopic = "dwm_unique_visit_2022";
        SingleOutputStreamOperator<VisitorStats> uvStatsDS = env
                .addSource(MyKafkaUtil.getKafkaSource(uvTopic, groupId))
                .map(new MapFunction<String, VisitorStats>() {
                    @Override
                    public VisitorStats map(String value) throws Exception {
                        JSONObject jsonObject = JSON.parseObject(value);
                        JSONObject common = jsonObject.getJSONObject("common");
                        VisitorStats visitorStats = new VisitorStats(
                                "",
                                "",
                                common.getString("vc"),
                                common.getString("ch"),
                                common.getString("ar"),
                                common.getString("is_new"),
                                1L, 0L, 0L, 0L, 0L, jsonObject.getLong("ts")
                        );
                        return visitorStats;
                    }
                });
        //TODO 5.从kafka读取dwm_user_jump_detail_2022主题数据、转化为VisitorStats pojo类、分配Watermark和提取事件时间戳
        String ujdTopic = "dwm_user_jump_detail_2022";
        SingleOutputStreamOperator<VisitorStats> ujdStatsDS = env
                .addSource(MyKafkaUtil.getKafkaSource(ujdTopic, groupId))
                .map(new MapFunction<String, VisitorStats>() {
                    @Override
                    public VisitorStats map(String value) throws Exception {
                        JSONObject jsonObject = JSON.parseObject(value);
                        JSONObject common = jsonObject.getJSONObject("common");
                        VisitorStats visitorStats = new VisitorStats(
                                "",
                                "",
                                common.getString("vc"),
                                common.getString("ch"),
                                common.getString("ar"),
                                common.getString("is_new"),
                                0L, 0L, 0L, 1L, 0L, jsonObject.getLong("ts")
                        );
                        return visitorStats;
                    }
                });

        //pageStatsDS.print();
        //TODO 6.使用union合并三条流条流,分组、开窗计算、聚合计算
        SingleOutputStreamOperator<VisitorStats> resultDS = pageStatsDS
                .union(uvStatsDS, ujdStatsDS)
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<VisitorStats>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner(new SerializableTimestampAssigner<VisitorStats>() {
                                    @Override
                                    public long extractTimestamp(VisitorStats element, long recordTimestamp) {
                                        return element.getTs();
                                    }
                                })
                )
                .keyBy(new KeySelector<VisitorStats, Tuple4<String, String, String, String>>() {
                    @Override
                    public Tuple4<String, String, String, String> getKey(VisitorStats value) throws Exception {
                        return Tuple4.of(
                                value.getAr(),
                                value.getCh(),
                                value.getIs_new(),
                                value.getVc()
                        );
                    }
                })
                .window(TumblingEventTimeWindows.of(Time.seconds(10L)))
                .reduce(
                        new ReduceFunction<VisitorStats>() {
                            @Override
                            public VisitorStats reduce(VisitorStats value1, VisitorStats value2) throws Exception {
                                value1.setPv_ct(value1.getPv_ct() + value2.getPv_ct());
                                value1.setSv_ct(value1.getSv_ct() + value2.getSv_ct());
                                value1.setUj_ct(value1.getUj_ct() + value2.getUj_ct());
                                value1.setUv_ct(value1.getUv_ct() + value2.getUv_ct());
                                value1.setDur_sum(value1.getDur_sum() + value2.getDur_sum());
                                return value1;
                            }
                        }
                        , new WindowFunction<VisitorStats, VisitorStats, Tuple4<String, String, String, String>, TimeWindow>() {
                            @Override
                            public void apply(Tuple4<String, String, String, String> tuple4, TimeWindow window, Iterable<VisitorStats> input, Collector<VisitorStats> out) throws Exception {
                                VisitorStats visitorStats = input.iterator().next();
                                visitorStats.setStt(DateTimeUtil.toYMDhms(new Date(window.getStart())));
                                visitorStats.setEdt(DateTimeUtil.toYMDhms(new Date(window.getEnd())));
                                out.collect(visitorStats);
                            }
                        });

        //TODO 7.将聚合结果写入Clickhouse
        resultDS
                .addSink(ClickHouseUtil.getSinkFunction("insert into  visitor_stats_2022 values(?,?,?,?,?,?,?,?,?,?,?,?)"));

        resultDS.print();
        //TODO 8.执行任务
        env.execute("VisitorStatsApp");
    }
}
